package com.gzsxy.kafka;

import com.alibaba.fastjson.JSON;
import com.gzsxy.entity.Order;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

/**
 * @author xiaolong
 * @version 1.0
 * @description: 生产者同步阻塞方法
 * @date 2022/7/5 21:00
 */
public class MySimpleProducer {

    private final static String TOPIC_NAME = "test";

    public static void main(String[] args) throws ExecutionException,InterruptedException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.113.229.173:9092");



        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //发消息的客户端
        Producer<String, String> producer = new KafkaProducer<String, String>(props);



        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME,
                    "mykey","hello" );


        //等待消息发送成功的同步阻塞方法
        RecordMetadata metadata = producer.send(producerRecord).get();
        //=====阻塞=======
        System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" +
                metadata.partition() + "|offset-" + metadata.offset());

        }




}
